package com.swak.consistency.reliablemsg.schedule;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;

import com.swak.consistency.reliablemsg.config.ReliablemsgConfig;
import com.swak.consistency.reliablemsg.context.Phase;
import com.swak.consistency.reliablemsg.context.ReliableMsgContext;
import com.swak.consistency.reliablemsg.context.ReliableMsgLog;
import com.swak.consistency.reliablemsg.handle.ReliablemsgConsistencyHandler;
import com.swak.consistency.reliablemsg.storage.ReliableMsgLogStorage;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.threads.ScheduledContext;

/**
 * 定时任务 - 定时处理异常的事件
 * 
 * @author DELL
 */
public class ReliablemsgScheduler implements InitializingBean, ApplicationListener<ApplicationReadyEvent> {
	protected Logger logger = LoggerFactory.getLogger(ReliablemsgScheduler.class);
	private ScheduledContext scheduledContext;
	private ReliablemsgConfig config;
	@Autowired
	private ReliableMsgLogStorage transLogStorage;
	@Autowired
	private ReliablemsgConsistencyHandler consistencyHandler;

	public ReliablemsgScheduler(ReliablemsgConfig config) {
		this.config = config;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		this.scheduledContext = Contexts.createScheduledContext("Reliablemsg-Scheduler-", 1, true, 60,
				TimeUnit.SECONDS);
	}

	@Override
	public void onApplicationEvent(ApplicationReadyEvent event) {
		this.scheduledContext.scheduleWithFixedDelay(() -> {
			try {
				this.doTask();
			} catch (Exception e) {
				logger.error("ReliablemsgScheduler Error:{}", e);
			}
		}, config.getTimeout() * 2, config.getTimeout(), TimeUnit.SECONDS);
	}

	private void doTask() {
		Date delay = new Date(System.currentTimeMillis() - config.getTimeout() * 1000);
		List<ReliableMsgLog> transLogs = this.transLogStorage.queryDelayMessage(delay);
		for (ReliableMsgLog trans : transLogs) {
			this.doHandle(trans);
		}
	}

	public void doHandle(ReliableMsgLog trans) {
		try {
			ReliableMsgContext context = new ReliableMsgContext();
			context.update(trans, Phase.valueOf(trans.getPhase()));
			context.execute(() -> {
				this.consistencyHandler.retry(context);
				return null;
			});
		} catch (Throwable e) {
			if (trans.getRetryTimes() >= trans.getMaxRts()) {
				logger.error("处理事务日志次数超过最大重试次数，需要人工介入处理，TransLog:{}", trans);
			}
			this.transLogStorage.retry(trans, trans.getMaxRts());
		}
	}
}